package com.etlmaster.executor.steps.zookeeper;

import java.io.IOException;
import java.util.List;

import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;

import com.etlmaster.executor.Step;
import com.etlmaster.executor.StepMeta;
import com.etlmaster.executor.annotation.StepInstruction;
import com.etlmaster.executor.scheduler.ScheduleServer;
import com.etlmaster.executor.scheduler.utils.ZkUtils;
import com.etlmaster.executor.utils.DataCollectionUtils;
import com.etlmaster.executor.utils.StepState;

/**
 * @author Larry
 * @copyright Larry@etlMaster
 * @since 2017-09-10
 */
@StepInstruction("watchZk")
public class WatchZkNodeStep extends Step {

	private WatchZkNodeStepMeta meta;

	private ZooKeeper zk;
	private Watcher wh;

	public WatchZkNodeStep(StepMeta meta) {
		super(meta);
		this.meta = (WatchZkNodeStepMeta) meta;

	}

	@Override
	protected int handle() {

		try {
			wh = ZkUtils.getWatcher();
			zk = ZkUtils.getInstance(
					DataCollectionUtils.getProperties("zookeeper.connect"),
					DataCollectionUtils.getIntProperties("zktimeout"), wh);

			zk.exists(meta.getZkPath(), true);

			List<String> children = zk.getChildren(meta.getZkPath(), true);
			if (children.size() > 0) {
				meta.log("INFO", "{}目录下子节点有:", meta.getZkPath());
				for (String child : children) {
					meta.log("INFO", child);
				}
			}
			meta.getStepState().getReturnValue().put("events", ScheduleServer.getEventMaps());

		} catch (IOException e) {
			meta.log("ERROR", e.getMessage());
			e.printStackTrace();
			return StepState.RUNNING_FAILED;
		} catch (KeeperException e) {
			e.printStackTrace();
			return StepState.RUNNING_FAILED;

		} catch (InterruptedException e) {
			e.printStackTrace();
			return StepState.RUNNING_FAILED;

		}
		return StepState.RUNNING_SUCCESSED;
	}

	@Override
	public boolean stop() {
		// TODO Auto-generated method stub
		return false;
	}

}
